package com.hjzd.ai.smartpractice.websocket;

import com.alibaba.nls.client.protocol.InputFormatEnum;
import com.alibaba.nls.client.protocol.NlsClient;
import com.alibaba.nls.client.protocol.SampleRateEnum;
import com.alibaba.nls.client.protocol.asr.SpeechRecognizer;
import com.alibaba.nls.client.protocol.asr.SpeechRecognizerListener;
import com.alibaba.nls.client.protocol.asr.SpeechRecognizerResponse;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.LongSerializationPolicy;
import com.google.gson.reflect.TypeToken;
import com.hjzd.ai.smartpractice.config.WebClientConfig;
import com.hjzd.ai.smartpractice.constant.LLMRole;
import com.hjzd.ai.smartpractice.constant.ServiceConstant;
import com.hjzd.ai.smartpractice.entity.DTO.chat.ChatCompletionMessage;
import com.hjzd.ai.smartpractice.entity.DTO.chat.ChatResponseThread;
import com.hjzd.ai.smartpractice.service.RedisService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.netty.http.client.HttpClient;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.hjzd.ai.smartpractice.constant.ServiceConstant.*;
import static com.hjzd.ai.smartpractice.websocket.SessionManager.*;

public class AliyunAsrClient {
    private static final Logger logger = LoggerFactory.getLogger(AliyunAsrClient.class);
    private RedisService redisService;
    private final NlsClient client;
    private SpeechRecognizer recognizer;
    private final WebSocketSession session;
    private final String appKey;
    private final AtomicBoolean isOpen = new AtomicBoolean(false);
    private WebClient webClient;
    private static final Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss")
            .setLongSerializationPolicy(LongSerializationPolicy.STRING).disableHtmlEscaping().create();
    private WebClientConfig webClientConfig;
    private ChatResponseThread ttsThread;
    private String ttsUrl;
    private String ttsType;
    private String llmFlag;
    private String saveVoicePath;
    private StepfunTTSClient ttsClient;
    private AliTtsClient aliTtsClient;
    public AliyunAsrClient(String appKey, WebSocketSession session, RedisService redisService,
                           WebClientConfig webClientConfig, String ttsUrl, String saveVoicePath, String ttsType
            , String llmFlag, String llmUrl) {
        this.appKey = appKey;
        this.session = session;
        this.redisService = redisService;
        this.webClientConfig = webClientConfig;
        this.ttsUrl = ttsUrl;
        this.ttsType = ttsType;
        this.saveVoicePath = saveVoicePath;
        this.client = new NlsClient("4cb2fe265d4147a18dee994741bda10a"); // 硬编码Token
        logger.info("阿里云ASR客户端已创建（使用硬编码Token）");
        HttpClient httpClient = HttpClient.create();
        ClientHttpConnector connector = new ReactorClientHttpConnector(httpClient);
        this.webClient = webClientConfig.webClientBuilder()
                .clientConnector(connector)
                .defaultHeader(HttpHeaders.CONTENT_TYPE, "application/json")
                .build();
        try {
            URI uri = new URI("wss://api.stepfun.com/v1/realtime/audio?model=step-tts-mini");
            this.ttsClient = new StepfunTTSClient(uri, session);
            this.ttsClient.connect();
        } catch (Exception e) {
            logger.error("初始化 Stepfun TTS 客户端失败", e);
        }
    }

    public void connect() {
        try {
            SpeechRecognizerListener listener = new SpeechRecognizerListener() {
                @Override
                public void onRecognitionResultChanged(SpeechRecognizerResponse response) {
                    String text = response.getRecognizedText();
                    //SessionManager.getPracticeCache(session).put(ASR_INTERMEDIATE_CACHE, text);
                    logger.debug("ASR中间结果: {}", text);
                }

                @Override
                public void onRecognitionCompleted(SpeechRecognizerResponse response) {
                    String text = response.getRecognizedText();
                    //tringBuilder asrResult = (StringBuilder) SessionManager.getPracticeCache(session).get(ASR_CACHE);
                    //asrResult.append(text);
                    try {
                        session.sendMessage(new TextMessage(inputTextStart));
                        session.sendMessage(new TextMessage(ASR_PREFIX + text));
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                    logger.info("ASR最终结果: {}", text);
                    handleFinalResult(session, text);
                }

                @Override
                public void onStarted(SpeechRecognizerResponse response) {
                    isOpen.set(true);
                    //logger.info("ASR连接已建立，taskId: {}", response.getTaskId());
                }

                @Override
                public void onFail(SpeechRecognizerResponse response) {
                    //logger.error("ASR识别失败: taskId={}, status={}, statusText={}", response.getTaskId(), response.getStatus(), response.getStatusText());
                    isOpen.set(false);
                    if (!isOpen.get()) {
                        //logger.warn("ASR连接未就绪，尝试重连...");
                        connect(); // 自动重连
                        if (!isOpen.get()) return;
                    }
                }
            };

            recognizer = new SpeechRecognizer(client, listener);
            recognizer.setAppKey(appKey);
            recognizer.setFormat(InputFormatEnum.PCM);
            recognizer.setSampleRate(SampleRateEnum.SAMPLE_RATE_16K);
            recognizer.setEnableIntermediateResult(true);
            recognizer.addCustomedParam("enable_voice_detection", true);
            recognizer.addCustomedParam("enable_punctuation_prediction", true);
            recognizer.start();

        } catch (Exception e) {
            logger.error("阿里云ASR连接失败", e);
            throw new RuntimeException("阿里云ASR连接失败", e);
        }
    }

    public void sendBytes(byte[] audioData) {
        if (!isOpen.get()) {
            logger.warn("ASR连接未就绪，忽略音频数据");
            return;
        }
        try {
            recognizer.send(audioData, audioData.length);
            // 根据16K采样率计算延迟
            Thread.sleep((audioData.length * 10 * 8000) / (160 * 16000));
        } catch (Exception e) {
            logger.error("发送音频数据失败", e);
        }
    }

    public void sendEof() {
        if (recognizer != null) {
            try {
                recognizer.stop();
            } catch (Exception e) {
                logger.error("发送EOF失败", e);
            }
        }
    }

    public void close() {
        try {
            if (recognizer != null) {
                recognizer.close();
            }
            client.shutdown();
            isOpen.set(false);
        } catch (Exception e) {
            logger.error("关闭ASR客户端失败", e);
        }
    }

    public boolean isOpen() {
        return isOpen.get();
    }


    private void handleFinalResult(WebSocketSession session, String text) {
        try {
            logger.info("ASR识别结果:" + text);

            // 1. 发送ASR结果到页面-用户自己说的
            //session.sendMessage(new TextMessage(text));
            // 使用打字机效果发送文本 100ms 延迟
            //sendTypewriterEffect(session, text, 100);
            session.sendMessage(new TextMessage(inputTextStop));
            logger.info("发送ASR消息 inputTextStop");

            // 2. 更新对话上下文
            String sessionId = SessionManager.getSession(session).getNick();
            List<ChatCompletionMessage> history = loadChatHistoryInline(sessionId);

            ChatCompletionMessage userMessage = new ChatCompletionMessage();
            userMessage.setContent(text);
            userMessage.setRole(LLMRole.USER.getRole());
            history.add(userMessage);
            logger.info("更新对话上下文");
            // 3. 调用大模型服务
            Flux<String> responseStream;
            ObjectNode requestBody;

            boolean stream = true; // 启用流式响应
            responseStream = webClient.post()
                    .uri("https://ark.cn-beijing.volces.com/api/v3/chat/completions")
                    .header(HttpHeaders.AUTHORIZATION, "Bearer 631999ff-4583-4944-9511-da68d2a58911")
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(buildVolcEngineRequest(session, text, stream))
                    .retrieve()
                    .bodyToFlux(String.class);
            session.sendMessage(new TextMessage(replyTextStart));

            logger.info("发送TTS replyTextStart");
            // 4. 处理流式响应
            ttsThread = new ChatResponseThread(session, this.ttsUrl, this.saveVoicePath, this.ttsType, this.ttsClient, this.aliTtsClient);
            ttsThread.start();

            responseStream.subscribe(
                    data -> {
                        try {
                            if ("[DONE]".equals(data)) {
                                // 向前端发送文本停止消息
                                session.sendMessage(new TextMessage(replyTextStop));
                                logger.info("发送TTS replyTextStop");
                                handleStreamCompletion(session, ttsThread);
                                ttsThread.sendFinalTextToTTS("。");
                                return;
                            }
                            ChatCompletionChunkResponse response = GSON.fromJson(data, ChatCompletionChunkResponse.class);
                            //logger.info("LLM原始响应:"+response.toString());
                            if (response != null &&
                                    response.getChoices() != null &&
                                    !response.getChoices().isEmpty() &&
                                    response.getChoices().get(0).getDelta() != null) {

                                //String content = response.getChoices().get(0).getDelta().getContent();
                                String content = response.getChoices().getFirst().getDelta().getContent().replace("\"", "");
                                if (content != null && !content.isEmpty()) {
                                    if (true) {
                                        ttsThread.sendTextToTTS(content);
                                        StringBuilder interfereCache = (StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE);
                                        interfereCache.append(content);
                                    } else {
                                        session.sendMessage(new TextMessage(TTS_PREFIX + content));
                                        StringBuilder interfereCache = (StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE);
                                        interfereCache.append(content);
                                        String tempTextCache = (String) SessionManager.getPracticeCache(session).get(TEMP_TEXT_CACHE);
                                        tempTextCache = tempTextCache + content;
                                        ConcurrentLinkedQueue<String> textQueue =
                                                (ConcurrentLinkedQueue<String>) SessionManager.getPracticeCache(session).get(SENTENCE_CACHE);
                                        //logger.info("[NOT DONE] tempTextCache：{}", tempTextCache);
                                        tempTextCache = sentenceSplit(SENTENCE_SPLIT_REGEX, tempTextCache, textQueue);
                                        //logger.info("[NOT DONE] sentenceCache：{}, tempTextCache:{}", textQueue, tempTextCache);
                                        SessionManager.getPracticeCache(session).put(TEMP_TEXT_CACHE, tempTextCache);
                                        //logger.info("subscribe content:{}, interfereCache:{}, textQueue:{}", content, interfereCache, textQueue);
                                    }

                                }
                            }

                        } catch (Exception e) {
                            logger.error("流数据处理异常: {}", e.getMessage());
                        }
                    },
                    error -> {
                        logger.error("流请求失败: {}", error.getMessage());
                        ttsThread.setStop();
                    },
                    () -> {
                        try {
                            session.sendMessage(new TextMessage(replyTextStop));
                            session.sendMessage(new TextMessage(replyAudioStop));
                        } catch (IOException e) {
                            logger.error("结束消息发送失败: {}", e.getMessage());
                        }
                    }
            );

        } catch (Exception e) {
            logger.error("最终结果处理失败: {}", e.getMessage());
            try {
                session.sendMessage(new TextMessage("ASR处理错误:" + e.getMessage()));
            } catch (IOException ioEx) {
                logger.error("错误消息发送失败: {}", ioEx.getMessage());
            }
        }
    }

    // 复用原有方法
    private String sentenceSplit(String regex, String text, ConcurrentLinkedQueue<String> queue) {
        Pattern pattern = Pattern.compile(regex);
        Matcher matcher = pattern.matcher(text);
        int lastPos = 0;
        StringBuilder buffer = new StringBuilder();

        while (matcher.find()) {
            buffer.append(text, lastPos, matcher.end());
            lastPos = matcher.end();

            if (buffer.length() > SENTENCE_SEND_MIN_WORDS_LIMIT) {
                queue.add(buffer.toString());
                buffer.setLength(0);
            }
        }
        return buffer.append(text.substring(lastPos)).toString();
    }


    // 复用原有逻辑处理完成状态
    private void handleStreamCompletion(WebSocketSession session, ChatResponseThread ttsThread) {
        try {
            // 保存助理响应
            String sessionId = SessionManager.getSession(session).getNick();
            List<ChatCompletionMessage> history = GSON.fromJson(
                    redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId)),
                    new TypeToken<List<ChatCompletionMessage>>() {
                    }.getType()
            );
            history.add(new ChatCompletionMessage(
                    LLMRole.ASSISTANT,
                    ((StringBuilder) SessionManager.getPracticeCache(session).get(INTERFERE_CACHE)).toString()
            ));
            redisService.set(
                    String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId),
                    GSON.toJson(history),
                    7, TimeUnit.DAYS
            );

            // 处理残留文本
            String tempText = (String) SessionManager.getPracticeCache(session).get(TEMP_TEXT_CACHE);
            if (StringUtils.isNotEmpty(tempText)) {
                ((ConcurrentLinkedQueue<String>) SessionManager.getPracticeCache(session).get(SENTENCE_CACHE)).add(tempText);
            }
            SessionManager.getPracticeCache(session).put(TEMP_TEXT_CACHE, "");

            // 停止TTS线程
            ttsThread.setStop();
        } catch (Exception e) {
            logger.error("流完成处理异常: {}", e.getMessage());
        }
    }

    private List<ChatCompletionMessage> loadChatHistoryInline(String sessionId) {
        String historyJson = redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId));
        return historyJson != null ?
                GSON.fromJson(historyJson, new TypeToken<List<ChatCompletionMessage>>() {
                }.getType()) :
                new ArrayList<>();
    }


    private ObjectNode buildVolcEngineRequest(WebSocketSession session, String text, boolean stream) {
        ObjectMapper objectMapper = new ObjectMapper();
        ObjectNode request = objectMapper.createObjectNode();
        request.put("model", "ep-20250117095237-vzjjf"); // 可以配置为动态参数
        request.put("stream", stream); // 动态设置流式响应

        ArrayNode messages = objectMapper.createArrayNode();

        // 内联历史记录加载
        String sessionId = SessionManager.getSession(session).getNick();
        List<ChatCompletionMessage> history = GSON.fromJson(
                redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, sessionId)),
                new TypeToken<List<ChatCompletionMessage>>() {
                }.getType()
        );
        if (history == null) {
            history = new ArrayList<>();
        }

        // 添加历史消息
        for (ChatCompletionMessage msg : history) {
            ObjectNode message = objectMapper.createObjectNode();
            String role = msg.getRole();
            if (role == null || role.trim().isEmpty()) {
                // 默认角色为 "user"
                role = LLMRole.USER.getRole();
            }
            message.put("role", role.toLowerCase());
            message.put("content", msg.getContent());
            messages.add(message);
        }

        // 添加当前消息
        ObjectNode currentMessage = objectMapper.createObjectNode();
        currentMessage.put("role", LLMRole.USER.getRole().toLowerCase());
        currentMessage.put("content", text);
        messages.add(currentMessage);

        // 校验messages是否为空
        if (messages.size() == 0) {
            throw new IllegalArgumentException("Messages cannot be empty");
        }

        request.set("messages", messages);
        return request;
    }


}
